//
// Created by whz on 2022/5/22.
//

#include "buffer_pool.h"
#include "../common/stop_watch.h"
#include "../enum//page_type.h"
#include <fcntl.h>
#include <unistd.h>
#include <cstring>

#if defined(__WIN32__)
//是否开启内存映射 1开0关
#define MMAP 0
#else
//是否开启内存映射 1开0关
#define MMAP 1

#include <sys/mman.h>

#endif

std::shared_ptr<BufferPool> BufferPool::instance = nullptr;
FlushQueue<PAGE *> BufferPool::flush_queue;

BufferPool::BufferPool() {
    load_checkpoint();
    for (int i = 0; i < conf.cursor; ++i) {
        std::cout << conf.free_array[i] << std::endl;
        free_list.push_back(conf.free_array[i]);
    }
    bool commit_status = check_file();
    if (!commit_status) {
        init_empty();
    }
    if (MMAP) {
        open_mmap();
    }
    //TODO 此处之后可优化，优先存储索引页（当count大于几倍pool_size以后..)
    //预热
    for (int i = 0; i < conf.count && i < POOL_SIZE; ++i) {
        bool is_free_page = (std::find(free_list.begin(), free_list.end(), i) != free_list.end());
        //非空闲页
        if (!is_free_page) {
            allocate(i);
        }
    }
    flush_service = std::make_shared<FlushService>();
    flush_service->start();
}

void BufferPool::init_empty() {
    //TODO There may be many info pages that need to be initialized
    PAGE *root = new PAGE();
    root->setPageNo(conf.count++);
    root->file_header.setFilPageType(FIL_PAGE_INDEX);
    deserialized(root, root->getPageNo());
}

BufferPool::~BufferPool() {
    if (MMAP) {
        close_mmap();
    }
}

PAGE *BufferPool::allocate(size_t page_no, bool read_only) {
    PAGE *new_page;
    if (page_no >= 0 && page_no < conf.count) {
        auto item = page_table.find(page_no);
        //内存命中
        if (item != page_table.end()) {
            new_page = item->second;
        } else {
//            std::cout << "不在内存" << std::endl;
            new_page = new PAGE();
            //不在内存在磁盘
            serialized(new_page, page_no);
            check_fill();
            page_table[new_page->getPageNo()] = new_page;
            page_list.push_front(new_page);
        }

    }

    if (page_no < 0 || page_no >= conf.count) {
        new_page = new PAGE();
//        std::cout << "申请新页" << std::endl;
        new_page->setPageNo(acquire());
        check_fill();
        page_table[new_page->getPageNo()] = new_page;
        page_list.push_front(new_page);
    }

    if (!read_only) {
        flush_set.insert(new_page);
    }

    return new_page;

}

size_t BufferPool::acquire() {
    if (free_list.empty()) {
        return conf.count++;
    }
    size_t free_page = free_list.front();
    free_list.pop_front();
//    std::cout << "从free_list中获取page[" << free_page << "]" << std::endl;
    return free_page;
}

void BufferPool::release(PAGE *page) {
    free_list.push_back((*page).getPageNo());
    page_table.erase((*page).getPageNo());
    page_list.remove(page);
    delete page;
}

void BufferPool::check_fill() {
    if (page_list.size() >= POOL_SIZE) {
        PAGE *page;
        for (;;) {
            page = page_list.back();
            page_list.pop_back();
            auto item = lock_set.find(page);
            if (page->getPageNo() == conf.root_no || item != lock_set.end()) {
                page_list.push_front(page);
            } else {
                break;
            }
        }
        page_table.erase((*page).getPageNo());
        flush(page);
    }
}


void BufferPool::flush(PAGE *page) {
    auto item = flush_set.find(page);
    if (item != flush_set.end()) {
//        std::cout<<"flush"<<page->getPageNo()<<std::endl;
        flush_queue.push(page);
        flush_set.erase(page);
    }
}

bool BufferPool::check_file(const std::string &path) {
    std::ifstream check(path);
    if (!check.good()) {
        std::fstream f(path, std::ios::out);
        f.close();
    }
    check.close();
    return check.good();
}

void BufferPool::serialized(PAGE *block, size_t page_no) {
    StopWatch stopWatch;
    off_t offset = page_no * sizeof(PAGE);
    if (MMAP) {
        read_mmap(block, offset);
    } else {
        f_read(block, offset);
    }
//    std::cout << "serialized page[" << page_no << "]:" << stopWatch.getElapsedTime() << "us" << std::endl;
}

void BufferPool::deserialized(PAGE *block, size_t page_no) {
    StopWatch stopWatch;
    off_t offset = page_no * sizeof(PAGE);
    f_write(block, offset);

//    std::cout << "deserialized page[" << page_no << "]:" << stopWatch.getElapsedTime() << "us" << std::endl;
}

void BufferPool::load_checkpoint() const {
    bool exist = check_file(PATH_CHECKPOINT);
    if (!exist) {
        return;
    }
    StopWatch stopWatch;

    std::ifstream fin(PATH_CHECKPOINT, std::ios::in | std::ios::binary);
    fin.read((char *) &conf, sizeof(checkpoint));
    fin.close();

//    std::cout << "load_checkpoint:" << stopWatch.getElapsedTime() << "us,next page:" << conf.count << std::endl;
}

void BufferPool::dump_checkpoint() const {
    StopWatch stopWatch;

    std::ofstream fout(PATH_CHECKPOINT, std::ios::out | std::ios::binary);
    fout.write((char *) &conf, sizeof(checkpoint));
    fout.close();

    std::cout << "dump_checkpoint:" << stopWatch.getElapsedTime() << "us,next page:" << conf.count << std::endl;

}

PAGE *BufferPool::get_root() {
    return allocate(conf.root_no);
}

void BufferPool::destroy() {
    std::cout << "destroy bufferPool" << std::endl;

    //stop flush service
    for (auto item: page_list) {
        flush_set.insert(item);
    }
    if (!flush_set.empty()) {
        for (auto item: flush_set) {
            flush_queue.push(item);
        }
    }
    flush_service->stop();

    //dump checkpoint
    conf.cursor = free_list.size();
    for (int i = 0; i < conf.cursor; ++i) {
        conf.free_array[i] = free_list.front();
        free_list.pop_front();
    }
    dump_checkpoint();


}

void BufferPool::set_root(PAGE *page) {
    conf.root_no = page->getPageNo();
}

void BufferPool::unlock(PAGE *page) {
    lock_set.erase(page);
}

void BufferPool::lock(PAGE *page) {
    lock_set.insert(page);
}

void BufferPool::f_read(PAGE *block, off_t offset) {
    FILE *fp = std::fopen(PATH_COMMIT, "rb+");
    fseek(fp, offset, SEEK_SET);
    fread(block, sizeof(PAGE), 1, fp);
    fclose(fp);
}

void BufferPool::f_write(PAGE *block, off_t offset) {
    FILE *fp = std::fopen(PATH_COMMIT, "rb+");
    fseek(fp, offset, SEEK_SET);
    fwrite(block, sizeof(PAGE), 1, fp);
    fclose(fp);
}

void BufferPool::read_stream(PAGE *block, off_t offset) {
    std::ifstream fin(PATH_COMMIT, std::ios::in | std::ios::binary);
    fin.seekg(offset);
    fin.read((char *) block, sizeof(PAGE));
    fin.close();
}

void BufferPool::write_stream(PAGE *block, off_t offset) {
    std::ofstream fout(PATH_COMMIT, std::ios::in | std::ios::out | std::ios::binary);
    fout.seekp(offset);
    fout.write((char *) block, sizeof(PAGE));
    fout.close();
}

void BufferPool::open_mmap() {
#if defined(__WIN32__)
#else
    int fd = open(PATH_COMMIT, O_RDONLY);
    size_t len = lseek(fd, 0, SEEK_END);
    buf = (char *) mmap(nullptr, len, PROT_READ, MAP_SHARED, fd, 0);
    close(fd);
#endif
}

void BufferPool::read_mmap(PAGE *block, off_t offset) {
    memcpy(block, buf + offset, sizeof(PAGE));
}

void BufferPool::close_mmap() {
#if defined(__WIN32__)
#else
    int fd = open(PATH_COMMIT, O_RDONLY);
    size_t len = lseek(fd, 0, SEEK_END);
    munmap(buf, len);
    close(fd);
#endif
}
